
#include <sys/types.h>
#include <time.h>
#include <string.h>
#include <stdlib.h>
#include <sstream>
#include <iomanip>
#include <vector>
#include <set>

#include <boost/algorithm/string.hpp>
#include <boost/lexical_cast.hpp>
#include <boost/filesystem.hpp>
using boost::system::error_code;
using boost::system::system_category;
using boost::system::system_error;

#include "bocat/common/base64.hpp"

#include "oplog.hpp"


namespace	sakcab	{

#define	IFSP_CACHE_SIZE		32

const std::string	OpLog::OP_PUT( "P" );
const std::string	OpLog::OP_GET( "G" );
const std::string	OpLog::OP_DEL( "D" );


const std::string	OpLog::SLICE_BASENAME( "oplog" );

const u_int32_t     OpLog::MIN_SLICE_SIZE_THRESHOLD = 1024 * 1024 * 32;
const u_int32_t     OpLog::MAX_SLICE_SIZE_THRESHOLD = 1024 * 1024 * 1024;


#define	OPLOG_FILE_ID	"footid"
#define OPLOG_FILE_ID_MAXIMUM	100000000


///////////////////////////////////////////////////////////////////////
//

OpLog::OpLog( const std::string data_dir, u_int32_t file_rotate_size, int slim, const char* key_encode, const char* value_encode )
		:	_data_dir( data_dir ),
			_file_rotate_size( file_rotate_size ),
			_slim( slim ),
			_ofs_append(),
			_ifsp_cache( IFSP_CACHE_SIZE ),
			_index_iterator_p( NULL ),
			_key_encode( key_encode ),
			_value_encode( value_encode ),
			_key_encode_base64_b(false),
			_value_encode_base64_b(false)
{
	if ( 0 == _file_rotate_size ) {
		_file_rotate_size = MAX_SLICE_SIZE_THRESHOLD;
	}
	if ( _file_rotate_size < MIN_SLICE_SIZE_THRESHOLD ) {
		_file_rotate_size = MIN_SLICE_SIZE_THRESHOLD;
	}
	if ( _file_rotate_size > MAX_SLICE_SIZE_THRESHOLD ) {
		_file_rotate_size = MAX_SLICE_SIZE_THRESHOLD;
	}

	if ( 0 == strcmp("base64", key_encode) ) {
		_key_encode_base64_b = true;
	}
	if ( 0 == strcmp("base64", value_encode) ) {
		_value_encode_base64_b = true;
	}

	_slim_lowater = int( _slim * 0.9 );
	if ( _slim_lowater < 1 ) {
		_slim_lowater = 1;
	}
	_slimming = false;

	initialize();
}

OpLog::~OpLog()
{
	finalize();
}

///////////////////////////////////////////////////////////////////////
//

void
OpLog::initialize()
{
	_min_fileno = 0;
	_max_fileno = 0;

	scanDataDirectory();

}

void
OpLog::finalize()
{
	if (_ofs_append.is_open()) 
		_ofs_append.close();

	closeAllReadFiles();

	if (_index_iterator_p) {
		delete _index_iterator_p;
		_index_iterator_p = NULL;
	}
}

///////////////////////////////////////////////////////////////////////
//


int	
OpLog::log( const Timestamp& ts, const std::string& op, const std::string& kpin, const std::string& key, const std::string& value )
{
	std::ofstream&	ofs = openAppendFile();
	
	std::string	kpin_stored = bocat::common::base64_encode( kpin );
	std::string	key_stored = key;
	if ( _key_encode_base64_b ) {
		key_stored = bocat::common::base64_encode( key );
	}
	std::string	value_stored = value;
	if ( _value_encode_base64_b ) {
		value_stored = bocat::common::base64_encode( value );
	}

	struct timeval	tv = ts.timeval();
	ofs << tv.tv_sec << "\t" << tv.tv_usec << "\t" << op << "\t" << kpin_stored << "\t" << key_stored << "\t" << value_stored << std::endl;

	if (ofs.tellp() > _file_rotate_size ) {
		ofs.close();
	}

	return	0;
}


bool
OpLog::isClear( void )
{
	int	count = countFiles();

	return	(count == 0);
}


///////////////////////////////////////////////////////////////////////
//

std::ofstream&
OpLog::openAppendFile( void )
{
	if (_ofs_append.is_open()) {
		return _ofs_append;
	}

	// find the only APPEND file
	std::string	fn_append = getAppendFileName();
	_ofs_append.open( fn_append.c_str(), std::ios::app );
	if ( !_ofs_append ) {
		std::ostringstream	msg;
		msg << "fail to open append file " << fn_append;
		throw std::runtime_error( msg.str() );
	}

	return _ofs_append;
}


std::string
OpLog::getAppendFileName( void )
{
	boost::filesystem::path	data_path( _data_dir );
	
	if (_max_fileno > 0) {
		std::string	fn = getCurrentAppendFileName();

		boost::filesystem::path	append_file = data_path / fn;
		if (boost::filesystem::exists( append_file ) ) {
			if (boost::filesystem::file_size(append_file) < _file_rotate_size) {
				return append_file.string();
			}
		}
	}

	std::string	fn = getNextAppendFileName();
	boost::filesystem::path	append_file = data_path / fn;
	return append_file.string();
}


std::string
OpLog::getNextAppendFileName( void )
{
	++_max_fileno;
	if ( _max_fileno >= OPLOG_FILE_ID_MAXIMUM ) {
		_max_fileno = 1;
	}

	return getFileName( _max_fileno );
}

std::string
OpLog::getCurrentAppendFileName( void )
{
	setCurrentFileID( _max_fileno );

	return getFileName( _max_fileno );
}


std::string
OpLog::getFileName( u_int32_t fileno )
{
	std::ostringstream	fn;

	fn << SLICE_BASENAME << "." << std::setw(9) << std::setfill('0') << fileno;

	return fn.str();
}


///////////////////////////////////////////////////////////////////////
//

std::ifstream&  
OpLog::openReadFile( u_int32_t fileno )
{
	boost::mutex::scoped_lock	_scoped_lock( this->_ifsp_cache_mutex );

	ifsp_type	ifsp;

	if ( _ifsp_cache.exists( fileno ) ) {
		ifsp = _ifsp_cache.get( fileno );
		return *ifsp;
	}

	ifsp = new std::ifstream();
	std::string	fn = getReadFileName( fileno );
	ifsp->open( fn.c_str() );
	if ( !(ifsp->good()) ) {
		std::ostringstream	msg;
		msg << "fail to open file " << fn;
		throw std::runtime_error( msg.str() );
	}

	ifsp_type	ifsp_removed = _ifsp_cache.put( fileno, ifsp );
	if ( ifsp_removed ) {
		if (ifsp_removed->is_open()) {
			ifsp_removed->close();
		}

		delete ifsp_removed;
	}

	return *ifsp;
}


void
OpLog::closeAllReadFiles( void )
{
	boost::mutex::scoped_lock	_scoped_lock( this->_ifsp_cache_mutex );

	ifsp_cache_type::Key_List	_list( _ifsp_cache.get_all_keys() );
	for( ifsp_cache_type::Key_List_Iter liter = _list.begin(); liter != _list.end(); liter++ ) {
		ifsp_type	_ifsp = _ifsp_cache.get( *liter, false );

		if ( _ifsp->is_open() ) {
			_ifsp->close();
		}
		delete _ifsp ;
	}

	_ifsp_cache.clear();
}


std::string
OpLog::getReadFileName( u_int32_t fileno )
{
	boost::filesystem::path	data_path( _data_dir );
	
	std::string	fn = getFileName( fileno );
	boost::filesystem::path	file_path = data_path / fn;
	return file_path.string();
}


///////////////////////////////////////////////////////////////////////
//

int
OpLog::clearDataDirectory( void )
{
	if (_ofs_append.is_open()) 
		_ofs_append.close();

	closeAllReadFiles();


	boost::filesystem::path	data_path( _data_dir );

	if ( !boost::filesystem::exists( data_path ) ) {
		if ( !boost::filesystem::create_directories( data_path ) ) {
			throw boost::filesystem::filesystem_error( "fail to create directory", data_path, error_code(errno, system_category) );
		}
	}

	int	count_file_removed = 0;

	boost::filesystem::directory_iterator end_itr; // default construction yields past-the-end
	for ( boost::filesystem::directory_iterator itr( data_path );
		itr != end_itr;
		++itr ) {
		if (boost::filesystem::is_regular_file( itr->status() ) ) {
			std::string	bn = boost::filesystem::basename( itr->path() );
			if ( bn == SLICE_BASENAME ) {
				boost::filesystem::remove( itr->path() );

				++count_file_removed;
			}
		}
	}

	return	count_file_removed;
}


int
OpLog::clearDataDirectory( const char* backup_dir )
{
	if (_ofs_append.is_open()) 
		_ofs_append.close();

	closeAllReadFiles();


	boost::filesystem::path	data_path( _data_dir );
	if ( !boost::filesystem::exists( data_path ) ) {
		if ( !boost::filesystem::create_directories( data_path ) ) {
			throw boost::filesystem::filesystem_error( "fail to create directory", data_path, error_code(errno, system_category) );
		}
	}

	boost::filesystem::path	bak_data_path( backup_dir );
	if ( !boost::filesystem::exists( bak_data_path ) ) {
		if ( !boost::filesystem::create_directories( bak_data_path ) ) {
			throw boost::filesystem::filesystem_error( "fail to create directory", bak_data_path, error_code(errno, system_category) );
		}
	}

	int	count_file_removed = 0;

	boost::filesystem::directory_iterator end_itr; // default construction yields past-the-end
	for ( boost::filesystem::directory_iterator itr( data_path );
		itr != end_itr;
		++itr ) {
		if (boost::filesystem::is_regular_file( itr->status() ) ) {
			std::string	bn = boost::filesystem::basename( itr->path() );
			if ( bn == SLICE_BASENAME ) {
				boost::filesystem::path	fn = itr->path().filename();
				boost::filesystem::path	path_to = bak_data_path / fn;

				boost::filesystem::rename( itr->path(), path_to );

				++count_file_removed;
			}
		}
	}

	return	count_file_removed;
}




///////////////////////////////////////////////////////////////////////
//

void
OpLog::scanDataDirectory( void )
{
	boost::filesystem::path	data_path( _data_dir );

	if ( !boost::filesystem::exists( data_path ) ) {
		if ( !boost::filesystem::create_directories( data_path ) ) {
			throw boost::filesystem::filesystem_error( "fail to create directory", data_path, error_code(errno, system_category) );
		}
	}

	boost::filesystem::path	file_id_path = data_path / OPLOG_FILE_ID;
	if ( !boost::filesystem::exists( file_id_path ) ) {
		int	rc = getCurrentFileID();
		if ( rc > 0 ) {
			_max_fileno = rc;
			return;
		}
	}
	
	boost::filesystem::directory_iterator end_itr; // default construction yields past-the-end
	for ( boost::filesystem::directory_iterator itr( data_path );
		itr != end_itr;
		++itr ) {
		if (boost::filesystem::is_regular_file( itr->status() ) ) {
			std::string	bn = boost::filesystem::basename( itr->path() );
			if ( bn == SLICE_BASENAME ) {
				std::string	en = boost::filesystem::extension( itr->path() );
				u_int32_t	fn = 0;
				try {
					fn = boost::lexical_cast< u_int32_t >( en.c_str() + 1 );	// skip '.'
				}
				catch( std::exception& e ) {
					continue;
				}

				if ( (0 == _min_fileno) || (_min_fileno > fn) ) {
						_min_fileno = fn;
				}
				if ( (0 == _max_fileno) || (_max_fileno < fn) ) {
						_max_fileno = fn;
				}
			}
		}
	}

}


int
OpLog::getCurrentFileID( void )
{
	boost::filesystem::path	data_path( _data_dir );

	boost::filesystem::path	file_id_path = data_path / OPLOG_FILE_ID;

	int	fileid = -1;
	std::ifstream	ifs;
	try {
		ifs.open( file_id_path.string().c_str() );
		std::string	line;
		std::getline( ifs, line );
		boost::trim( line );

		fileid = boost::lexical_cast< int >( line.c_str() );
	}
	catch( ... ) {
		// pass
	}

	if ( ifs.is_open() ) {
		ifs.close();
	}

	return	fileid;
}

int
OpLog::setCurrentFileID( uint32_t id )
{
	boost::filesystem::path	data_path( _data_dir );

	boost::filesystem::path	file_id_path = data_path / OPLOG_FILE_ID;

	std::ofstream	ofs;
	try {
		ofs.open( file_id_path.string().c_str() );
		ofs << id << std::endl;
	}
	catch( ... ) {
		// pass
	}

	if ( ofs.is_open() ) {
		ofs.close();
	}

	return	0;
}


///////////////////////////////////////////////////////////////////////
//

OpLogIterator*
OpLog::getSlimOpLogIterator( void )
{
	if (_slim < 1) {
		return	NULL;
	}

	if (!_slimming) {
		int	count = countFiles();
		if ( count < _slim ) {
			return	NULL;
		}

		_slimming = true;
	}

	boost::mutex::scoped_lock	_scoped_lock( this->_index_iterator_mutex );
	if ( NULL != _index_iterator_p ) {
		// already in sliming
		return	NULL;
	}

	std::string	file_to_slim;
	int	fileno_to_slim = getFileToSlim( file_to_slim ); 
	if ( fileno_to_slim > 0 ) {
		_file_slimming = file_to_slim;
		_fileno_slimming = fileno_to_slim;

		_index_iterator_p = new OpLogIterator( file_to_slim.c_str(), _key_encode, _value_encode );
		return _index_iterator_p;
	}
	else {
		// no more file to slim
		_slimming = false;
	}

	return	NULL;
}

void
OpLog::freeSlimOpLogIterator( OpLogIterator* it, bool done )
{
	if ( (NULL != it) && (it == _index_iterator_p) ) {
		delete _index_iterator_p;

		if (done && (_file_slimming.size() > 0)) {
			boost::filesystem::path	slim_file( _file_slimming );
			try {
				boost::filesystem::remove( slim_file );
			}
			catch( ... ) {
				// pass
			}
		}

		_index_iterator_p = NULL;
	}
}


int
OpLog::getFileToSlim( std::string& file_to_slim )
{
	if (_slim < 1) {
		return	0;
	}

	boost::filesystem::path	data_path( _data_dir );

	if ( !boost::filesystem::exists( data_path ) ) {
		if ( !boost::filesystem::create_directory( data_path ) ) {
			throw boost::filesystem::filesystem_error( "fail to create directory", data_path, error_code(errno, system_category) );
		}
	}

	int	count_files = 0;
	//u_int32_t	oldest_size = 0;
	std::time_t	oldest_mtime = time(NULL) + 86400;
	boost::filesystem::path	oldest_file;

	boost::filesystem::directory_iterator end_itr; // default construction yields past-the-end
	for ( boost::filesystem::directory_iterator itr( data_path );
		itr != end_itr;
		++itr ) {
		if (boost::filesystem::is_regular_file( itr->status() ) ) {
			std::string	bn = boost::filesystem::basename( itr->path() );
			if ( bn == SLICE_BASENAME ) {
				++count_files;

				//// oplog should ignore size of log file
				////   log file should be delete in first create/modify first delete order
				//u_int32_t	file_size = boost::filesystem::file_size( itr->path() );
				//if ( file_size < _file_rotate_size ) {
				//	// skip the last file
				//	continue;
				//}

				time_t	mtime = boost::filesystem::last_write_time( itr->path() );
				if (mtime < oldest_mtime) {
					oldest_mtime = mtime;
					oldest_file = itr->path();
				}
				//else if (mtime == oldest_mtime) {
				//	if ( file_size > oldest_size ) {
				//		oldest_size = file_size;
				//		oldest_file = itr->path();
				//	}
				//}
			}
		}
	}

	if (count_files <= _slim_lowater) {
		return	0;
	}

	file_to_slim = oldest_file.string();
	int	fileno = getFileNoFromName( file_to_slim );

	return	fileno;
}

int
OpLog::getFileNoFromName( const std::string& filename )
{
	boost::filesystem::path	path( filename );

	std::string	ext = boost::filesystem::extension( path );
	int	fileno = 0;
	try {
		// skip '.000001'
		fileno = boost::lexical_cast< int >( ext.c_str() + 1 );
	}
	catch( ... ) {
	}

	return	fileno;
}

int
OpLog::countFiles( void )
{
	int	count_files = 0;

	boost::filesystem::path	data_path( _data_dir );

	if ( !boost::filesystem::exists( data_path ) ) {
		if ( !boost::filesystem::create_directory( data_path ) ) {
			throw boost::filesystem::filesystem_error( "fail to create directory", data_path, error_code(errno, system_category) );
		}
	}

	boost::filesystem::directory_iterator end_itr; // default construction yields past-the-end
	for ( boost::filesystem::directory_iterator itr( data_path );
		itr != end_itr;
		++itr ) {
		if (boost::filesystem::is_regular_file( itr->status() ) ) {
			std::string	bn = boost::filesystem::basename( itr->path() );
			if ( bn == SLICE_BASENAME ) {
				++count_files;
			}
		}
	}

	return	count_files;
}


int
OpLog::getTankage( void )
{
	int	count_files = 0;
	uint32_t	tankage = 0;

	boost::filesystem::path	data_path( _data_dir );

	if ( !boost::filesystem::exists( data_path ) ) {
		if ( !boost::filesystem::create_directory( data_path ) ) {
			throw boost::filesystem::filesystem_error( "fail to create directory", data_path, error_code(errno, system_category) );
		}
	}

	boost::filesystem::directory_iterator end_itr; // default construction yields past-the-end
	for ( boost::filesystem::directory_iterator itr( data_path );
		itr != end_itr;
		++itr ) {
		if (boost::filesystem::is_regular_file( itr->status() ) ) {
			std::string	bn = boost::filesystem::basename( itr->path() );
			if ( bn == SLICE_BASENAME ) {
				++count_files;
				u_int32_t	file_size = boost::filesystem::file_size( itr->path() );
				tankage += file_size;

				if ( tankage >= 0x7FFFFFFF ) {
					tankage = 0x7FFFFFFF;
					break;
				}
			}
		}
	}

	return	static_cast< int >( tankage );
}


///////////////////////////////////////////////////////////////////////
//

void
OpLog::beginRecovery( void )
{
	std::set< std::string >	filenames;

	boost::filesystem::path	data_path( _data_dir );

	boost::filesystem::directory_iterator end_itr; // default construction yields past-the-end
	for ( boost::filesystem::directory_iterator itr( data_path );
		itr != end_itr;
		++itr ) {
		if (boost::filesystem::is_regular_file( itr->status() ) ) {
			std::string	bn = boost::filesystem::basename( itr->path() );
			if ( bn == SLICE_BASENAME ) {
				filenames.insert( itr->path().string() );
			}
		}
	}

	std::set< std::string >::iterator	it = filenames.begin();
	for ( ; it != filenames.end(); ++it ) {
		_filelist_recovery.push_back( *it );
	}

	_filelist_index_recovery = 0;
}


void
OpLog::endRecovery( void )
{
	_filelist_recovery.clear();
	_filelist_index_recovery = 0;
}


bool
OpLog::hasMoreRecoveryIterator( void )
{
	return ( _filelist_index_recovery < _filelist_recovery.size() );
}


OpLogIterator*
OpLog::getNextRecoveryIterator( void )
{
	if ( _filelist_index_recovery < _filelist_recovery.size() ) {
		const std::string&	file_to_slim = _filelist_recovery[ _filelist_index_recovery ];

		_index_iterator_p = new OpLogIterator( file_to_slim.c_str(), _key_encode, _value_encode );
		++_filelist_index_recovery;

		return _index_iterator_p;
	}

	return	NULL;
}


}	//// namespace sakcab

